package lib;

import java.lang.Thread;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.InetAddress;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.concurrent.LinkedBlockingQueue;

public class ListenSocket extends ServerSocket implements Runnable { 
	private Thread msgThread;
	private volatile LinkedBlockingQueue<Message> messages;
	private String actorType;
	private ActorAddress superActor;
	private ActorAddress myAddress;
	private boolean isAvailable;
	private static final int BACKLOG=10;
	
	public ListenSocket(int port,String actorType, ActorAddress superActor, ActorAddress myAddress) throws Exception {
		super(port);
		this.actorType=actorType;
		this.superActor=superActor;
		this.myAddress=myAddress;
		this.isAvailable=true;
		messages = new LinkedBlockingQueue<Message>();
		msgThread = new Thread(this);
		msgThread.start();
	}
	
	public void run(){
		ObjectInputStream  objIn;
		Socket client;
		boolean continues = true;

		try {
			do {
				client = this.accept();
				objIn = new ObjectInputStream(client.getInputStream());
				Message msg = (Message) objIn.readObject();
				messages.put(msg);
				client.close();
				if (msg.getOperation().equals("Terminate") && (msg.getDestActorType().equals(this.actorType)) && (msg.getSrcActorType().equals("SuperActor"))){
					continues=false;
				}
				//Informe automatiquement le superacteur en cas de grosse charge sur notre acteur
				if (messages.size()>=BACKLOG){
					if (isAvailable!=false){
						Message msgToSend=new Message(myAddress,actorType,"SuperActor","Update my state",null);
						Socket conn=superActor.getSocket();
						ObjectOutputStream objOut=new ObjectOutputStream(conn.getOutputStream());
						objOut.writeObject(msgToSend);
						conn.close();
						isAvailable=false;
					}
				}
			} while (continues);
			this.close();
		}
		catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	public Message getMessage(){
		Message msg = messages.poll();
		//Informe automatiquement le superacteur lorsque la charge diminue sur notre acteur
		try{
			if (messages.size()<BACKLOG) {
				if (isAvailable!=true){
					Message msgToSend=new Message(myAddress,actorType,"SuperActor","Update my state",null);
					Socket conn=superActor.getSocket();
					ObjectOutputStream objOut=new ObjectOutputStream(conn.getOutputStream());
					objOut.writeObject(msgToSend);
					conn.close();
					isAvailable=true;
				}
			}
		}
		catch (Exception e) {
			e.printStackTrace();
			
		}
		return(msg);
	}
	
	public Message getMessageBlocking(){
		Message msg = null;
		try{
			msg = messages.take();//Informe automatiquement le superacteur lorsque la charge diminue sur notre acteur
			if (messages.size()<BACKLOG) {
				if (isAvailable!=true){
					Message msgToSend=new Message(myAddress,actorType,"SuperActor","Update my state",null);
					Socket conn=superActor.getSocket();
					ObjectOutputStream objOut=new ObjectOutputStream(conn.getOutputStream());
					objOut.writeObject(msgToSend);
					conn.close();
					isAvailable=true;
				}
			}
		}
		catch (Exception e) {
			e.printStackTrace();
			
		}
		return(msg);
	}
}
